package org.redis.cluster.transport.client.support;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.transport.client.Client;
import org.redis.cluster.transport.client.ClientPool;
import org.redis.cluster.transport.client.IPooledClientFactory;
import org.redis.cluster.transport.protocol.ProtocolDecoder;
import org.redis.cluster.transport.protocol.ProtocolEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultClientPool implements ClientPool,Plugin {
	
	private final static String DEFAULT_PROTOCOL_DECODER="clientDecoder";
	
	private final static String DEFAULT_PROTOCOL_ENCODER="clientEncoder";
	
	private final static String DEFAULT_POOLED_CLIENT_FACTORY="pooledClientFactory";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private final ConcurrentHashMap<ClientConfig, GenericObjectPool<Client>> poolMap=new ConcurrentHashMap<ClientConfig, GenericObjectPool<Client>>();

	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	private ProtocolDecoder decoder;
	private ProtocolEncoder encoder;

	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		
		this.decoder=initDecoder();
		this.encoder=initEncoder();
	}
	
	@Override
	public void destroy() {
		if(MapUtils.isNotEmpty(poolMap)) {
			return;
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("准备断开后端连接");
		}
		
		for(Entry<ClientConfig,GenericObjectPool<Client>> entry:poolMap.entrySet()) {
			if(logger.isDebugEnabled()) {
				logger.debug("准备断开后端连接 - {}",entry.getKey());
			}
			
			entry.getValue().close();
			
			if(logger.isDebugEnabled()) {
				logger.debug("断开后端连接完成 - {}",entry.getKey());
			}
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("断开后端连接完成");
		}
	}
	
	@Override
	public Client getClient(ClientConfig config) {
		GenericObjectPool<Client> pool=getClientPool(config);
		try {
			return pool.borrowObject();
		}
		catch(Exception e) {
			if(logger.isDebugEnabled()) {
				logger.debug("",e);
			}
			
			return null;
		}
	}
	
	@Override
	public boolean returnClient(ClientConfig config, Client client) {
		GenericObjectPool<Client> pool=getClientPool(config);
		pool.returnObject(client);
		return true;
	}
	
	@Override
	public boolean destroy(ClientConfig config) {
		GenericObjectPool<Client> pool=getClientPool(config);
		if(pool==null) {
			return false;
		}

		removeClientPool(config);
		pool.close();
		return true;
	}
	
	protected GenericObjectPool<Client> getClientPool(ClientConfig config) {
		if(poolMap.containsKey(config)) {
			return poolMap.get(config);
		}
		
		GenericObjectPoolConfig poolConfig=getPoolConfig(config);
		IPooledClientFactory pooledClientFactory=getPoolFactory(config);
		
		GenericObjectPool<Client> pool=initClientPool(poolConfig, pooledClientFactory);
		poolMap.putIfAbsent(config, pool);
		return getClientPool(config);
	}
	
	protected boolean removeClientPool(ClientConfig config) {
		return poolMap.remove(config)!=null;
	}

	protected GenericObjectPool<Client> initClientPool(GenericObjectPoolConfig poolConfig,IPooledClientFactory pooledClientFactory) {
		return new GenericObjectPool<Client>(pooledClientFactory,poolConfig);
	}

	protected IPooledClientFactory getPoolFactory(ClientConfig config) {
		return extensionLoader.load(DEFAULT_POOLED_CLIENT_FACTORY, IPooledClientFactory.class, config, decoder,encoder);
	}
	
	protected GenericObjectPoolConfig getPoolConfig(ClientConfig config) {
		GenericObjectPoolConfig poolConfig=new GenericObjectPoolConfig();
		poolConfig.setMaxTotal(config.getConnection());
		return poolConfig;
	}
	
	protected ProtocolDecoder initDecoder() {
		return extensionLoader.load(DEFAULT_PROTOCOL_DECODER, ProtocolDecoder.class);
	}
	
	protected ProtocolEncoder initEncoder() {
		return extensionLoader.load(DEFAULT_PROTOCOL_ENCODER, ProtocolEncoder.class);
	}
}